import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 读取文件，按主键 写入 MySQL
 */
@Slf4j
public class SqlDemo1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //创建source表
        String sql = "" +
                "CREATE TABLE dwd_customer_da (\n" +
                "  id int," +
                "  name String,"+
                "  `date` String," +
                "  account bigint" +
                ") WITH (\n" +
                "  'connector'='filesystem',\n" +
                "  'path'='file/dwd_customer_da/',\n" +
                "  'format'='csv'\n" +
                ")";


        String sql2 = "" +
                "CREATE TABLE fs_sink (\n" +
                "  user_id STRING," +
                "  age int"+
                ") WITH (\n" +
                "  'connector'='filesystem',\n" +
                "  'path'='file/sink',\n" +
                "  'format'='csv'\n" +
                ")";
        String sinkSql3 = "CREATE TABLE dws_customer_da ( id int, name String, account bigint, PRIMARY KEY (id) NOT ENFORCED) WITH (  'connector' = 'jdbc',  'url' = 'jdbc:mysql://localhost:3306/test?useSSL=false',  'table-name' = 'dws_customer_da',  'username' = 'root',  'password' = '123456')";
        tableEnv.executeSql(sql);
        tableEnv.executeSql(sinkSql3);

        System.out.println(sql);
        System.out.println(sinkSql3);

        // sum 值一致在变，所以必须使用 toRetractStream
        /*Table tableResult = tableEnv.sqlQuery("select id, name, sum(account) from dwd_customer_da group by id, name");
        tableEnv.toRetractStream(tableResult, Row.class).print();
        env.execute();*/

        // 按主键更新保存到 mysql
        // Table sink 'default_catalog.default_database.fs_sink' doesn't support consuming update changes which is produced by node GroupAggregate(groupBy=[user_id], select=[user_id, SUM(age) AS EXPR$1])
        StatementSet stmtSet = tableEnv.createStatementSet();
        stmtSet.addInsertSql("insert into dws_customer_da select id, name, sum(account) from dwd_customer_da group by id, name");
        stmtSet.execute();

        // 写入数据到 mysql
        /*Table tableResult = tableEnv.sqlQuery("select user_id, sum(age) from mysql_sink group by user_id");
        tableEnv.toRetractStream(tableResult, Row.class).print();

        env.execute();*/
    }
}
